package pers.vic.mq.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.listener.MessageListener;
import org.springframework.kafka.listener.MessageListenerContainer;

import com.alibaba.fastjson.JSONObject;

import pers.vic.mq.kafka.model.EventMsg;

/**
 * 配置一些KAFKA消费者 测试用 
 * 
 * @author VIC
 *
 */
@Configuration
@EnableKafka
public class TestConsumeConfig extends BaseConsumeConfig{
	
	//测试主题1
	@Value("${kafka.test.one.topic}")
	private String topic1;
	
	//测试主题2
	@Value("${kafka.test.two.topic}")
	private String topic2;
	
	//test 消费者组1
	@Value("${kafka.test.one.group}")
	private String group1;		
	
	//test 消费者组1
	@Value("${kafka.test.two.group}")
	private String group2;		
			
	
	@Bean(name = { "testConsumer1" })
	public MessageListenerContainer createTestConsumer() {
		return createConsumer(group1, topic1, 1, new MessageListener<String, String>() {
			public void onMessage(ConsumerRecord<String, String> record) {
				String value = record.value();
				EventMsg obj = JSONObject.parseObject(value, EventMsg.class);
				logger.info("-----------group1   for  topic1---------EventMsg：{}---", obj.toString());
			}
		});
	}

	@Bean(name = { "testConsumer2" })
	public MessageListenerContainer createOtherConsumer() {
		

		return createConsumer(group1, topic2, 1, new MessageListener<String, String>() {
			public void onMessage(ConsumerRecord<String, String> record) {
				String value = record.value();
				EventMsg obj = JSONObject.parseObject(value, EventMsg.class);
				logger.info("-----------group1   for  topic2---------EventMsg：{}---", obj.toString());
			}
		});
	}
	
	@Bean(name = { "testConsumer3" })
	public MessageListenerContainer createOtherConsumer2() {
		

		return createConsumer(group2, topic2, 1, new MessageListener<String, String>() {
			public void onMessage(ConsumerRecord<String, String> record) {
				String value = record.value();
				EventMsg obj = JSONObject.parseObject(value, EventMsg.class);
				logger.info("-----------group2   for  topic2---------EventMsg：{}---", obj.toString());
			}
		});
	}


}
